package com.mini.grpc.examples.advanced;

import com.mini.grpc.client.ManagedChannel;
import com.mini.grpc.client.ManagedChannelBuilder;
import com.mini.grpc.common.Deadline;
import com.mini.grpc.loadbalance.LoadBalancer;
import com.mini.grpc.loadbalance.LoadBalancerRegistry;
import com.mini.grpc.retry.ExponentialBackoffRetryPolicy;
import com.mini.grpc.retry.RetryPolicy;
import com.mini.grpc.stream.StreamObserver;

import java.net.InetSocketAddress;
import java.util.Arrays;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.logging.Logger;

/**
 * 高级功能示例
 * 展示负载均衡、重试机制和流式调用的使用
 * 
 * @author Mini-gRPC
 */
public class AdvancedFeaturesExample {
    
    private static final Logger logger = Logger.getLogger(AdvancedFeaturesExample.class.getName());
    
    public static void main(String[] args) throws Exception {
        // 演示负载均衡
        demonstrateLoadBalancing();
        
        // 演示重试机制
        demonstrateRetryPolicy();
        
        // 演示超时机制
        demonstrateDeadline();
        
        // 演示流式调用
        demonstrateStreamingCalls();
    }
    
    /**
     * 演示负载均衡功能
     */
    private static void demonstrateLoadBalancing() {
        logger.info("=== 负载均衡演示 ===");
        
        // 创建负载均衡器
        LoadBalancerRegistry registry = LoadBalancerRegistry.getInstance();
        
        // 轮询负载均衡器
        LoadBalancer roundRobinLB = registry.newLoadBalancer("round_robin", Arrays.asList(
                new InetSocketAddress("server1.example.com", 9090),
                new InetSocketAddress("server2.example.com", 9090),
                new InetSocketAddress("server3.example.com", 9090)
        ));
        
        logger.info("轮询负载均衡器: " + roundRobinLB.getName());
        for (int i = 0; i < 6; i++) {
            logger.info("第" + (i + 1) + "次选择: " + roundRobinLB.choose());
        }
        
        // 选择第一个负载均衡器
        LoadBalancer pickFirstLB = registry.newLoadBalancer("pick_first", Arrays.asList(
                new InetSocketAddress("server1.example.com", 9090),
                new InetSocketAddress("server2.example.com", 9090),
                new InetSocketAddress("server3.example.com", 9090)
        ));
        
        logger.info("\n选择第一个负载均衡器: " + pickFirstLB.getName());
        for (int i = 0; i < 3; i++) {
            logger.info("第" + (i + 1) + "次选择: " + pickFirstLB.choose());
        }
        
        // 关闭负载均衡器
        roundRobinLB.shutdown();
        pickFirstLB.shutdown();
    }
    
    /**
     * 演示重试策略
     */
    private static void demonstrateRetryPolicy() {
        logger.info("\n=== 重试策略演示 ===");
        
        // 创建指数退避重试策略
        RetryPolicy retryPolicy = ExponentialBackoffRetryPolicy.newBuilder()
                .setMaxAttempts(5)
                .setInitialDelayMillis(100)
                .setMaxDelayMillis(5000)
                .setMultiplier(2.0)
                .setJitter(0.1)
                .build();
        
        logger.info("重试策略: " + retryPolicy.getName());
        logger.info("最大重试次数: " + retryPolicy.getMaxAttempts());
        
        // 模拟重试延迟计算
        for (int attempt = 1; attempt <= retryPolicy.getMaxAttempts(); attempt++) {
            long delay = retryPolicy.getRetryDelayMillis(attempt);
            logger.info("第" + attempt + "次重试延迟: " + delay + "ms");
        }
    }
    
    /**
     * 演示超时机制
     */
    private static void demonstrateDeadline() {
        logger.info("\n=== 超时机制演示 ===");
        
        // 创建5秒超时
        Deadline deadline = Deadline.after(5, TimeUnit.SECONDS);
        logger.info("创建超时: " + deadline);
        
        // 检查剩余时间
        try {
            Thread.sleep(1000); // 等待1秒
            logger.info("1秒后剩余时间: " + deadline.timeRemaining(TimeUnit.MILLISECONDS) + "ms");
            
            Thread.sleep(2000); // 再等待2秒
            logger.info("3秒后剩余时间: " + deadline.timeRemaining(TimeUnit.MILLISECONDS) + "ms");
            
            logger.info("是否过期: " + deadline.isExpired());
            
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
    
    /**
     * 演示流式调用
     */
    private static void demonstrateStreamingCalls() throws InterruptedException {
        logger.info("\n=== 流式调用演示 ===");
        
        // 演示客户端流式调用
        demonstrateClientStreaming();
        
        // 演示服务端流式调用
        demonstrateServerStreaming();
        
        // 演示双向流式调用
        demonstrateBidiStreaming();
    }
    
    /**
     * 演示客户端流式调用
     */
    private static void demonstrateClientStreaming() throws InterruptedException {
        logger.info("--- 客户端流式调用 ---");
        
        CountDownLatch latch = new CountDownLatch(1);
        
        // 响应观察者
        StreamObserver<String> responseObserver = new StreamObserver<String>() {
            @Override
            public void onNext(String value) {
                logger.info("收到响应: " + value);
            }
            
            @Override
            public void onError(Throwable t) {
                logger.severe("发生错误: " + t.getMessage());
                latch.countDown();
            }
            
            @Override
            public void onCompleted() {
                logger.info("客户端流式调用完成");
                latch.countDown();
            }
        };
        
        // 模拟请求流
        StreamObserver<String> requestObserver = new StreamObserver<String>() {
            @Override
            public void onNext(String value) {
                logger.info("发送请求: " + value);
            }
            
            @Override
            public void onError(Throwable t) {
                logger.severe("请求流错误: " + t.getMessage());
            }
            
            @Override
            public void onCompleted() {
                logger.info("请求流完成");
                // 模拟处理完成后发送响应
                responseObserver.onNext("处理了多个请求的结果");
                responseObserver.onCompleted();
            }
        };
        
        // 发送多个请求
        requestObserver.onNext("请求1");
        requestObserver.onNext("请求2");
        requestObserver.onNext("请求3");
        requestObserver.onCompleted();
        
        latch.await(5, TimeUnit.SECONDS);
    }
    
    /**
     * 演示服务端流式调用
     */
    private static void demonstrateServerStreaming() throws InterruptedException {
        logger.info("--- 服务端流式调用 ---");
        
        CountDownLatch latch = new CountDownLatch(1);
        
        // 响应观察者
        StreamObserver<String> responseObserver = new StreamObserver<String>() {
            private int count = 0;
            
            @Override
            public void onNext(String value) {
                logger.info("收到响应 " + (++count) + ": " + value);
            }
            
            @Override
            public void onError(Throwable t) {
                logger.severe("发生错误: " + t.getMessage());
                latch.countDown();
            }
            
            @Override
            public void onCompleted() {
                logger.info("服务端流式调用完成，共收到 " + count + " 个响应");
                latch.countDown();
            }
        };
        
        // 模拟服务端发送多个响应
        String request = "获取数据列表";
        logger.info("发送请求: " + request);
        
        // 模拟服务端处理
        new Thread(() -> {
            try {
                for (int i = 1; i <= 5; i++) {
                    Thread.sleep(200); // 模拟处理时间
                    responseObserver.onNext("数据项 " + i);
                }
                responseObserver.onCompleted();
            } catch (InterruptedException e) {
                responseObserver.onError(e);
            }
        }).start();
        
        latch.await(5, TimeUnit.SECONDS);
    }
    
    /**
     * 演示双向流式调用
     */
    private static void demonstrateBidiStreaming() throws InterruptedException {
        logger.info("--- 双向流式调用 ---");
        
        CountDownLatch latch = new CountDownLatch(1);
        
        // 响应观察者
        StreamObserver<String> responseObserver = new StreamObserver<String>() {
            private int responseCount = 0;
            
            @Override
            public void onNext(String value) {
                logger.info("收到响应: " + value);
                responseCount++;
            }
            
            @Override
            public void onError(Throwable t) {
                logger.severe("发生错误: " + t.getMessage());
                latch.countDown();
            }
            
            @Override
            public void onCompleted() {
                logger.info("双向流式调用完成，共收到 " + responseCount + " 个响应");
                latch.countDown();
            }
        };
        
        // 请求观察者
        StreamObserver<String> requestObserver = new StreamObserver<String>() {
            @Override
            public void onNext(String value) {
                logger.info("发送请求: " + value);
                // 模拟服务端立即响应
                responseObserver.onNext("回复: " + value);
            }
            
            @Override
            public void onError(Throwable t) {
                logger.severe("请求流错误: " + t.getMessage());
                responseObserver.onError(t);
            }
            
            @Override
            public void onCompleted() {
                logger.info("请求流完成");
                responseObserver.onCompleted();
            }
        };
        
        // 发送多个请求
        new Thread(() -> {
            try {
                for (int i = 1; i <= 3; i++) {
                    Thread.sleep(300); // 模拟发送间隔
                    requestObserver.onNext("消息 " + i);
                }
                requestObserver.onCompleted();
            } catch (InterruptedException e) {
                requestObserver.onError(e);
            }
        }).start();
        
        latch.await(5, TimeUnit.SECONDS);
    }
} 